package com.borened.netty.protocol.mqtt;

import com.borened.netty.util.SpringUtil;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.*;

/**
 * @author chengcaihua
 * @description 订阅者管理器
 * @since 2023-11-13 16:59
 */
@Slf4j
@Component
public class SubscriberManager {

    private final Set<Subscriber> subscribers = new HashSet<>();

    private final ExecutorService executorService = Executors.newFixedThreadPool(10, r -> new Thread(r, "mqtt-subscriber"));

    public void executeEvent(Runnable subscribeEvent){
        executorService.submit(()->{
            try {
                subscribeEvent.run();
            } catch (Exception e) {
                log.error("执行订阅事件失败,错误详情：",e);
            }
        });
    }
    public void addSubscriber(Subscriber subscriber) {
        subscribers.add(subscriber);
        log.info("添加订阅成功，topic:{},clientId:{}",subscriber.getTopic(),subscriber.getMqttClientSession().getClientId());
        triggerSubscribeEvent(subscriber);
    }
    public void removeSubscriber(Subscriber subscriber) {
        subscribers.remove(subscriber);
        log.info("移除订阅成功，topic:{},clientId:{}",subscriber.getTopic(),subscriber.getMqttClientSession().getClientId());
    }
    /**
     * 触发订阅事件
     */
    private void triggerSubscribeEvent(Subscriber subscriber) {
        executeEvent(()-> {
            MqttPublisher publisher = SpringUtil.getBean(MqttPublisher.class);
            Map<String, List<MqttPersistentMessage>> retainedMessages = publisher.getRetainedMessages();
            //todo topic filter mode
            if (retainedMessages.containsKey(subscriber.getTopic())){
                retainedMessages.get(subscriber.getTopic()).forEach(subscriber::receiveMessage);
            }
        });
    }

    /**
     * 匹配订阅者
     * @param topic 匹配主题
     * @return
     */
    public Set<Subscriber> getSubscribers(String topic) {
        Set<Subscriber> filters = new HashSet<>();
        for (Subscriber subscriber : subscribers) {
            //todo topic filter mode
            if (subscriber.getTopic().equals(topic)) {
                filters.add(subscriber) ;
            }
        }
        return filters;
    }
}
